package com.diao.flink.streamapi.transform

import com.diao.flink.streamapi.source.SensorReading
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
/**
 * @author: Chenzhidiao
 * @date: 2020/3/26 16:30
 * @description:滚动聚合算子的应用
 * @version: 1.0
 */
object Test1 {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val stream: DataStream[SensorReading] = env.readTextFile("E:\\Workspace\\flink\\src\\main\\resources\\sensorFile").map(data=>{
      //将每条数据按逗号切分，然后封装成SensorReading对象
      val dataArray = data.split(",")
      SensorReading(dataArray(0).trim,dataArray(1).trim.toLong,dataArray(2).trim.toDouble)
    })
    //滚动聚合算子应用
    val data: DataStream[SensorReading] = stream.keyBy(0).min("temperature")

    data.print()
    env.execute()
  }

}
